package cn.mzy.xhr;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

//消息发送者
public class Consumer {

    public static void main(String[] args) {
        try {
            // 实例化消息生产者Producer
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumergroup");
            // 设置NameServer的地址
            consumer.setNamesrvAddr("127.0.0.1:9876");
            // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
            consumer.subscribe("xhr_order_topic", "xhr_pet_tag");
            // 注册回调实现类来处理从broker拉取回来的消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    try {
                        for (MessageExt msg : msgs) {
                            // msg.getBody():拿到的是消息是一个byte数组
                            System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8));
                        }
                        // 标记该消息已经被成功消费
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            });
            // 启动Producer实例
            consumer.start();
            System.out.println("Consumer Started.%n");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}